package com.kamistoat.meimeimall.meimeimallpayment.service.impl;

import com.kamistoat.meimeimall.common.constant.DatabaseConstant;
import com.kamistoat.meimeimall.common.constant.PaymentConstant;
import com.kamistoat.meimeimall.common.constant.SecKillConstant;
import com.kamistoat.meimeimall.common.exception.BizCodeEnum;
import com.kamistoat.meimeimall.common.exception.RRException;
import com.kamistoat.meimeimall.meimeimallpayment.dao.BillRecordDao;
import com.kamistoat.meimeimall.meimeimallpayment.entity.BillRecordEntity;
import com.kamistoat.meimeimall.meimeimallpayment.entity.ReconciliationTaskEntity;
import com.kamistoat.meimeimall.meimeimallpayment.service.BillRecordService;
import com.kamistoat.meimeimall.meimeimallpayment.service.PaymentRecordService;
import org.apache.ibatis.session.SqlSessionFactory;
import org.redisson.api.RSemaphore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

@Service("billRecordService")
public class BillRecordServiceImpl implements BillRecordService {
    protected Logger logger = LoggerFactory.getLogger(BillRecordServiceImpl.class);

    @Qualifier("flushBillThreadPoolExecutor")
    @Autowired
    ThreadPoolExecutor flushBillThreadPoolExecutor;

    @Autowired
    BillRecordDao billRecordDao;

    /**
     * 为指定日期内的第三方对账cvs文件创建数据库表，并执行数据清洗
     *
     * @param reconciliationTask 对账任务
     * @return 创建的数据库表名
     */
    @Transactional(rollbackFor = RRException.class)
    public String flushTradeTable(ReconciliationTaskEntity reconciliationTask) {
        String billTableName = PaymentConstant.BILL_RECORD_PREFIX + reconciliationTask.getId();

        /**
         * 建表
         */
        try {
            logger.info("建立第三方账单临时表{}", billTableName);
            billRecordDao.createTableWithName(billTableName);
        } catch (Exception exception) {
            throw new RRException(BizCodeEnum.CREATE_BILL_TABLE_ERROR.getMessage(), BizCodeEnum.CREATE_BILL_TABLE_ERROR.getCode());
        }

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        LocalDate startDay = LocalDate.parse(
                sdf.format(reconciliationTask.getStartTime()).split(" ")[0],
                DateTimeFormatter.ofPattern("yyyy-MM-dd"));
        LocalDate endDay = LocalDate.parse(
                sdf.format(reconciliationTask.getEndTime()).split(" ")[0],
                DateTimeFormatter.ofPattern("yyyy-MM-dd"));
        logger.info("清洗{}至{}的第三方账单，目标数据表：{}", startDay, endDay, billTableName);

        /**
         * 一个对账任务可能包含多天，扫描这些天的第三方账单
         */
        List<CompletableFuture<Void>> asyncList = new ArrayList<>();
        List<BillRecordEntity> totalBillRecordList = new ArrayList<>();
        while (startDay.isBefore(endDay)) {
            // csv文件名
            String fileName = "/home/kamistoat/ubuntu/LiGuStore/tradeSource/" + startDay.toString().replaceAll("-", "") + "_" + "业务明细.csv";
            String line = "";
            int billCsvSkipHeadline = PaymentConstant.BILL_CSV_SKIP_HEADLINE;
            try (BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(fileName), "GBK"))) {
                // 读文件
                while ((line = br.readLine()) != null) {
                    // 跳过#注解行
                    if (!line.startsWith("#")) {
                        // 跳过无效数据行
                        if (billCsvSkipHeadline == 0) {
                            String[] splits = line.split(PaymentConstant.BILL_CSV_SPLIT);
                            BillRecordEntity billRecordEntity = new BillRecordEntity();
                            billRecordEntity.setThirdPaymentTradeSn(splits[0]);
                            billRecordEntity.setOrderSn(splits[1]);
                            billRecordEntity.setPaymentType(splits[2].equals("付款") ?
                                    PaymentConstant.PaymentType.PAYMENT.getCode() :
                                    PaymentConstant.PaymentType.REFUND.getCode());
                            billRecordEntity.setTotalAmount(new BigDecimal(splits[3]));
                            totalBillRecordList.add(billRecordEntity);

                            /**
                             * 账单数达到单次清洗上限，开新线程将其写入数据库
                             */
                            if (totalBillRecordList.size() == PaymentConstant.BILL_FLUSH_SINGLE_LIMIT) {
                                // 数据拷贝要在异步任务开启之前执行
                                List<BillRecordEntity> listAsync = new ArrayList<>(totalBillRecordList);
                                totalBillRecordList.clear();
                                asyncList.add(
                                        CompletableFuture.runAsync(() -> {
                                            billRecordDao.insertBatchWithName(listAsync, billTableName);
                                            listAsync.clear();
                                        }, flushBillThreadPoolExecutor));
                            }
                        } else {
                            billCsvSkipHeadline--;
                        }
                    }
                }
            } catch (Exception exception) {
                throw new RRException(BizCodeEnum.READ_BILL_CSV_ERROR.getMessage() + ":" + exception, BizCodeEnum.READ_BILL_CSV_ERROR.getCode());
            }
            startDay = startDay.plusDays(1);
        }

        // 最后一轮的手动开新线程进行清洗
        List<BillRecordEntity> listAsync = new ArrayList<>(totalBillRecordList);
        totalBillRecordList.clear();
        asyncList.add(
                CompletableFuture.runAsync(() -> {
                    billRecordDao.insertBatchWithName(listAsync, billTableName);
                    listAsync.clear();
                }, flushBillThreadPoolExecutor));

        try {
            CompletableFuture.allOf(asyncList.toArray(new CompletableFuture[0])).get();
        } catch (Exception exception) {
            throw new RRException(BizCodeEnum.FLUSH_BILL_TO_MYSQL_ERROR.getMessage() + ":" + exception, BizCodeEnum.FLUSH_BILL_TO_MYSQL_ERROR.getCode());
        }

        logger.info("清洗{}至{}的第三方账单完成，目标数据表：{}", startDay, endDay, billTableName);
        return billTableName;
    }
}
